package com._51doit.rpc

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable
import scala.concurrent.duration._

/**
 * @create: 2021-10-08 19:12
 * @author: 珠穆朗玛峰顶峰的男人
 * @program: AKKaRpc
 **/
class Master extends Actor {
  val id2Worker = mutable.HashMap[String, WorkerInfo]()


  override def preStart(): Unit = {
    import context.dispatcher
    //定时给自己 发消息检测超时的worker
    context.system.scheduler.schedule(0 second, 15 second, self, CheckTimeOutWorker)
  }

  //接收消息的方法
  override def receive: Receive = {
    case RegisterWorker(workerId, memory, cores) => {
      //将worker信息保存起来
      val workerInfo = new WorkerInfo(workerId, memory, cores)
      //将workerInfo保存到map中
      id2Worker(workerId) = workerInfo
      //向worker发送一个注册成功的信息
      sender() ! RegisterWorker
    }

    case HeartBeat(workerId) => {
      //更新对应worker的最近一次心跳时间
      if (id2Worker.contains(workerId)) {
        val workerInfo: WorkerInfo = id2Worker(workerId)
        //将当前的系统时间赋给对应的workerinfo的最近心跳时间
        workerInfo.lastHeartBeatTime = System.currentTimeMillis()
      }
    }
    case CheckTimeOutWorker => {
      //检测超时的Worker
      val deadWorker: Iterable[WorkerInfo] = id2Worker.values.filter(w => System.currentTimeMillis() - w.lastHeartBeatTime > 15000)
      //将超时的Worker从map中移除
      deadWorker.foreach(w => id2Worker -= w.workerId)

      println(s"current alive worker size is : ${id2Worker.size}")
    }
  }
}

object Master {

  def main(args: Array[String]): Unit = {
    val configStr =
      """
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = "127.0.0.1"
        |akka.remote.netty.tcp.port = 8888
        |""".stripMargin;
    val config: Config = ConfigFactory.parseString(configStr)
    //创建ActorSystem
    val masterActorSystem: ActorSystem = ActorSystem("MasterActorSystem", config)
    //创建Actor
    val masterActor = masterActorSystem.actorOf(Props[Master], "MasterActor")

  }
}
